package com.zl.learning.reactive;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * @author zhaolei
 */
public class RxJavaTest2 {
    public static void main(String[] args) throws InterruptedException {
        Observable<String> girl = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
                System.out.println(Thread.currentThread().getName());
                emitter.onNext("1");
                Thread.sleep(1000);
                emitter.onNext("2");
                Thread.sleep(1000);
                emitter.onNext("3");
                Thread.sleep(1000);
                emitter.onNext("4");
                Thread.sleep(1000);
                emitter.onNext("5");
                Thread.sleep(1000);
                emitter.onComplete();
            }
        });
        Observer<String> man = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable disposable) {
                System.out.println("onSubscribe" + disposable);
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println(Thread.currentThread().getName());
                System.out.println("onNext" + s);
            }

            @Override
            public void onError(@NonNull Throwable throwable) {
                System.out.println("onError" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };
        girl.observeOn(Schedulers.computation()).subscribeOn(Schedulers.computation()).subscribe(man);
        Thread.sleep(10000);
    }
}
